package im.oen.boot.common.queue;

import im.oen.boot.common.constant.OenCode;
import im.oen.boot.common.utils.Checker;
import im.oen.boot.common.utils.DateUtil;
import im.oen.boot.common.utils.Runner;
import org.jetbrains.annotations.NotNull;

import java.io.Serializable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * 延迟队列执行器
 *
 * @author Kevin.Xu
 */
public class DelayQueueExecutor {

    static Logger logger = Logger.getLogger(DelayQueueExecutor.class.getName());

    private static ConcurrentHashMap<Serializable, Consumer<Serializable>> queues = new ConcurrentHashMap<>();
    private static DelayQueue delayQueueKeys = new DelayQueue();

    public static void run(Serializable key, Consumer<Serializable> consumer, long millis) {
        if (Checker.isEmpty(key) || Checker.isEmpty(consumer)) {
            throw OenCode.PARAMETER_IS_EMPTY.exception();
        }
        if (millis <= 0) {
            throw OenCode.PARAMETER_VALUE_RANGE_ERROR.exception();
        }
        delayQueueKeys.add(new DelayQueueKey(key, millis));
        queues.put(key, consumer);
    }

    public static void cancel(Serializable key) {
        queues.remove(key);
        logger.log(Level.INFO, "Cancel Delay Task : {0}", key);
    }

    private static void run() {
        Thread thread = new Thread(() -> {
            while (true) {
                Runner.safeRun(() -> {
                    DelayQueueKey delayQueueKey = (DelayQueueKey) DelayQueueExecutor.delayQueueKeys.take();
                    if (Checker.isNotEmpty(delayQueueKey)) {
                        Serializable key = delayQueueKey.getKey();
                        Consumer<Serializable> consumer = DelayQueueExecutor.queues.get(key);
                        if (Checker.isNotEmpty(consumer)) {
                            logger.log(Level.INFO, "Start Run Delay Task : {0}", key);
                            Runner.safeRun(consumer, key);
                        }
                    } else {
                        Thread.sleep(100L);
                    }
                });
            }
        });
        thread.setDaemon(true);
        thread.setName("DelayQueueExecutor-Thread");
        thread.start();
    }

    static {
        run();
    }

    private static class DelayQueueKey implements Delayed {
        private final Serializable key;
        private final long start;
        private final long millis;

        DelayQueueKey(Serializable key, long millis) {
            this.key = key;
            this.start = DateUtil.currentTimeMillis();
            this.millis = millis;
        }

        @Override
        public long getDelay(@NotNull TimeUnit unit) {
            return unit.convert((start + millis) - DateUtil.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(@NotNull Delayed delayed) {
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - delayed.getDelay(TimeUnit.MILLISECONDS));
        }

        public Serializable getKey() {
            return key;
        }
    }
}
